package day02

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 创建RDD
 *
 * 1、从集合中创建
 * 2、从外部存储系统的数据集创建
 * 3、从其他RDD创建
 */
object CreateRDD {}

/**
 * 1、从集合中创建
 */
object CreateRDDFROMArray {
  def main(args: Array[String]): Unit = {

    //1、创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("CreateRDDFROMArray").setMaster("local[*]")

    //2、创建SparkContext，该对象是提交Spark App 的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.1、使用parallelize()创建rdd
    val rdd1: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4, 5, 6))

    rdd1.foreach(println)

    // 获取此RDD的分区数组的长度
    val rdd1Partitions: Int = rdd1.partitions.length
    println("rdd1分区数：" + rdd1Partitions)

    println("--------------------------------")

    //3.2、使用makeRDD()创建rdd
    val rdd2: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5, 6, 7, 8))
    rdd2.foreach(println)

    val rdd2Partitions: Int = rdd2.partitions.length
    println("rdd2分区数：" + rdd2Partitions)

    // 关闭SparkContext
    sc.stop()
  }
}

/**
 * 2、从外部存储系统的数据集创建
 *
 * 由外部存储系统的数据集创建RDD包括：本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、HBase等
 */
object CreateRDDFROMOuter {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setAppName("CreateRDDFROMOuter").setMaster("local[2]")

    val sc: SparkContext = new SparkContext(conf)

    val rdd: RDD[String] = sc.textFile("input")
    rdd.foreach(println)

    sc.stop()
  }
}

/**
 * 从hdfs数据集创建
 */
object CreateRDDFROMHDFS {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setAppName("CreateRDDFROMHDFS").setMaster("local[2]")

    val sc: SparkContext = new SparkContext(conf)

    // hadoop3.X的web端口是9870、内部通讯端口是8020
    // hadoop2.X的web端口是50070、内部通讯端口是9000
    val rdd: RDD[String] = sc.textFile("hdfs://hadoop102:8020/input")
    rdd.foreach(println)
    rdd.saveAsTextFile("output")

    sc.stop()
  }
}

/**
 * 3、从其他RDD创建
 */
object CreateRDDFROMOTHERRDD {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setAppName("CreateRDDFROMOTHERRDD").setMaster("local[*]")

    val sc: SparkContext = new SparkContext(sparkConf)

    val rdd1: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5))

    val rdd2: RDD[(Int, Int)] = rdd1.map((_, 1))

    val tuples: Array[(Int, Int)] = rdd2.collect()

    tuples.foreach(println)
  }
}

